package mongodb

import (
	"context"
	"errors"
	"go-toolkit/library/log"
	"go-toolkit/library/net/trace"
	xtime "go-toolkit/library/time"
	"strings"
	"sync"
	"time"

	"go.mongodb.org/mongo-driver/event"
	"go.mongodb.org/mongo-driver/mongo"
	"go.mongodb.org/mongo-driver/mongo/options"
	"go.mongodb.org/mongo-driver/mongo/readpref"
	"go.mongodb.org/mongo-driver/x/mongo/driver/connstring"
)

const (
	slowDuration     = 250 // ms
	defaultBucketNum = 16
)

func (c *Monitor) Started(ctx context.Context, evt *event.CommandStartedEvent) {
	t, ok := trace.FromContext(ctx)
	if !ok {
		return
	}

	newSpan := t.Fork("", "mongodb")
	newSpan.SetTag(trace.SpanKindClientTag)
	newSpan.SetTag(trace.TagString(trace.TagComponent, "library/database/mongo"))
	newSpan.SetTag(trace.TagString("command", evt.CommandName))
	newSpan.SetTag(trace.TagString("databaseName", evt.DatabaseName))
	newSpan.SetTag(trace.TagString("connectionID", evt.ConnectionID))
	newSpan.SetTag(trace.TagInt64("requestID", evt.RequestID))
	container := c.buckets[bucketIndex(evt.RequestID)]
	container.locker.Lock()
	container.spans[spanKey{evt.ConnectionID, evt.RequestID}] = newSpan
	container.locker.Unlock()

}

func (c *Monitor) Succeeded(ctx context.Context, evt *event.CommandSucceededEvent) {
	container := c.buckets[bucketIndex(evt.RequestID)]
	spankey := spanKey{evt.ConnectionID, evt.RequestID}
	container.locker.Lock()
	span, ok := container.spans[spankey]
	if ok {
		delete(container.spans, spankey)
	}
	container.locker.Unlock()

	if ok {
		span.Finish(nil)
	}
	addr := addrFromConnectID(evt.ConnectionID)
	duration := time.Duration(evt.DurationNanos).Milliseconds()
	_metricReqDur.Observe(duration, addr, addr, evt.CommandName)
	if duration > slowDuration {
		log.Warnc(ctx, "mongo slow log duration:%d commandName:%s ", duration, evt.CommandName)
	}

}

func (c *Monitor) Faild(ctx context.Context, evt *event.CommandFailedEvent) {
	container := c.buckets[bucketIndex(evt.RequestID)]
	spankey := spanKey{evt.ConnectionID, evt.RequestID}
	container.locker.Lock()
	span, ok := container.spans[spankey]
	if ok {
		delete(container.spans, spankey)
	}
	container.locker.Unlock()
	if ok {
		err := errors.New(evt.Failure)
		span.Finish(&err)
	}

	addr := addrFromConnectID(evt.ConnectionID)
	duration := time.Duration(evt.DurationNanos).Milliseconds()

	_metricReqDur.Observe(duration, addr, addr, evt.CommandName)
	_metricReqErr.Inc(addr, addr, evt.CommandName, evt.Failure)

}

func addrFromConnectID(connectionID string) (addr string) {
	addr = connectionID
	if idx := strings.IndexByte(addr, '['); idx >= 0 {
		addr = addr[:idx]
	}
	return
}

func mongoAddr(connectionID string) (hostname, port string) {
	hostname = connectionID
	if idx := strings.IndexByte(hostname, '['); idx >= 0 {
		hostname = hostname[:idx]
	}
	if idx := strings.IndexByte(hostname, ':'); idx >= 0 {
		port = hostname[idx+1:]
		hostname = hostname[:idx]
	}
	return hostname, port
}

func bucketIndex(requestID int64) int64 {
	return requestID & (defaultBucketNum - 1)
}

type Monitor struct {
	buckets []*Container
}

type Container struct {
	locker sync.Mutex
	spans  map[spanKey]trace.Trace
}

type spanKey struct {
	ConnectionID string
	RequestID    int64
}

type Config struct {
	URI             string
	MaxPoolSize     uint64
	MinPoolSize     uint64
	ConnectTimeout  xtime.Duration
	MaxConnIdleTime xtime.Duration
	SocketTimeout   xtime.Duration
}

type Client struct {
	*mongo.Client
	db *mongo.Database
}

// DB get a default database that read as URI ReadPreference config
func (c *Client) DB() *mongo.Database {
	return c.db
}

func NewClient(conf *Config) (*Client, error) {
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	moni := &Monitor{
		buckets: make([]*Container, defaultBucketNum),
	}
	for i := 0; i < defaultBucketNum; i++ {
		moni.buckets[i] = &Container{spans: make(map[spanKey]trace.Trace)}
	}

	cmdMonitor := &event.CommandMonitor{
		Started:   moni.Started,
		Succeeded: moni.Succeeded,
		Failed:    moni.Faild,
	}
	if conf.MinPoolSize > conf.MaxPoolSize {
		return nil, errors.New("conf.MinPoolSize > conf.MaxPoolSize")
	}

	poolMonitor := &event.PoolMonitor{
		Event: func(evt *event.PoolEvent) {
			switch evt.Type {
			case event.GetSucceeded:
				_metricConnCurrent.Add(1, evt.Address, "active")
			case event.ConnectionReturned:
				_metricConnCurrent.Add(-1, evt.Address, "active")

			case event.ConnectionCreated:
				_metricConnTotal.Add(1, evt.Address)
			case event.ConnectionClosed:
				_metricConnTotal.Add(-1, evt.Address)
			}
		},
	}

	opt := options.Client().ApplyURI(conf.URI).SetMonitor(cmdMonitor).SetPoolMonitor(poolMonitor)
	if conf.ConnectTimeout > 0 {
		opt.SetConnectTimeout(time.Duration(conf.ConnectTimeout))
	}
	if conf.MaxConnIdleTime > 0 {
		opt.SetMaxConnIdleTime(time.Duration(conf.MaxConnIdleTime))
	}
	if conf.SocketTimeout > 0 {
		opt.SetSocketTimeout(time.Duration(conf.SocketTimeout))
	}
	if conf.MaxPoolSize > 0 {
		opt.SetMaxPoolSize(conf.MaxPoolSize)
	}
	if conf.MinPoolSize > 0 {
		opt.SetMinPoolSize(conf.MinPoolSize)
	}

	client, err := mongo.Connect(ctx, opt)
	if err != nil {
		return nil, err
	}
	// Ping the primary
	if err := client.Ping(ctx, readpref.Primary()); err != nil {
		return nil, err
	}
	cs, err := connstring.ParseAndValidate(conf.URI)
	if err != nil {
		return nil, err
	}

	cli := &Client{
		Client: client,
		db:     client.Database(cs.Database),
	}

	return cli, nil
}
